package com.wxzhou.wangzhuan.scheduler;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.wxzhou.wangzhuan.constants.LoginEvent;
import com.wxzhou.wangzhuan.repository.LoginEventRespository;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.stream.ReactiveStreamCommands;
import io.quarkus.redis.datasource.stream.StreamMessage;
import io.quarkus.redis.datasource.stream.XReadGroupArgs;
import io.quarkus.scheduler.Scheduled;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@ApplicationScoped
public class RedisLoginConsumer {

    private static final Logger LOG = Logger.getLogger(RedisLoginConsumer.class);
    private static final int BATCH_SIZE = 10;
    private static final String STREAM_KEY = "login:events";
    private static final String CONSUMER_GROUP = "login-consumers";
    private static final String CONSUMER_NAME = "consumer-2";

    private final ReactiveStreamCommands<String, String, String> streamCommands;

    private final ObjectMapper objectMapper;

    @Inject
    LoginEventRespository loginEventRespository;

    @Inject
    public RedisLoginConsumer(ReactiveRedisDataSource reactiveRedisDataSource, ObjectMapper objectMapper) {
        this.streamCommands = reactiveRedisDataSource.stream(String.class, String.class, String.class);
        this.objectMapper = objectMapper;
        // 初始化时创建消费者组
        createConsumerGroupIfNotExists();
    }

//    @Scheduled(every = "10s")
    public void consumeLoginEvents() {
        try {
            // 确保消费者组存在
            createConsumerGroupIfNotExists();

            // 使用 XREADGROUP 从消费者组读取消息
            streamCommands.xreadgroup(
                    CONSUMER_GROUP,
                    CONSUMER_NAME,
                    STREAM_KEY,
                    ">",
                    new XReadGroupArgs().count(BATCH_SIZE).block(Duration.ZERO)
            ).subscribe().with(
                    streamMessages -> {
                        LOG.infof("Received %d login events from Redis stream", streamMessages.size());
                        // 处理消息
                        processLoginEvents(streamMessages);
                    },
                    throwable -> {
                        LOG.error("Error reading login events from Redis stream", throwable);
                    }
            );
        } catch (Exception e) {
            LOG.error("Error consuming login events from Redis stream", e);
        }
    }

    private void processLoginEvents(List<StreamMessage<String, String, String>> records) {
        try {
            // 存储解析成功的登录事件和对应的消息ID
            List<LoginEvent> loginEvents = new ArrayList<>();
            Map<String, StreamMessage<String, String, String>> messageMap = new HashMap<>();

            // 解析所有消息
            for (StreamMessage<String, String, String> record : records) {
                // 检查 record 是否为 null
                if (record == null) {
                    LOG.warnf("Received null record in batch");
                    continue;
                }

                // 获取消息 ID 和数据
                String messageId = record.id();
                Map<String, String> eventData = record.payload();

                // 检查 eventData 是否为 null 或空
                if (eventData == null || eventData.isEmpty()) {
                    LOG.warnf("Received empty or null event data for message ID: %s", messageId);
                    continue;
                }

                // 检查必要字段是否存在
                if (!eventData.containsKey("data")) {
                    LOG.warnf("Missing required 'data' field in event data for message ID: %s", messageId);
                    continue;
                }

                try {
                    // 从事件数据中获取序列化的登录事件
                    String eventJson = eventData.get("data");
                    if (eventJson == null || eventJson.isEmpty()) {
                        LOG.warnf("Received empty or null event JSON for message ID: %s", messageId);
                        continue;
                    }

                    // 将事件数据反序列化为 LoginEvent 对象
                    LoginEvent loginEvent = objectMapper.readValue(eventJson, LoginEvent.class);

                    // 添加到列表中
                    loginEvents.add(loginEvent);
                    messageMap.put(messageId, record);
                } catch (Exception e) {
                    LOG.error("Error parsing login event with ID: %s", messageId, e);
                }
            }

            // 批量插入到数据库
            if (!loginEvents.isEmpty()) {
                loginEventRespository.batchInsertComments(loginEvents)
                        .subscribe().with(
                                rowCount -> {
                                    LOG.infof("Successfully inserted %d login events into database", rowCount);
                                    // 批量确认和删除消息
                                    acknowledgeAndDeleteMessages(messageMap, rowCount);
                                },
                                throwable -> {
                                    LOG.error("Error inserting login events into database", throwable);
                                    // 记录失败日志，不确认消息，这样消息会在下次继续被消费
                                    logFailedMessages(messageMap);
                                }
                        );
            }
        } catch (Exception e) {
            LOG.error("Error processing login events batch", e);
        }
    }

    private void acknowledgeAndDeleteMessages(Map<String, StreamMessage<String, String, String>> messageMap, int insertedRowCount) {
        try {
            // 使用AtomicInteger来包装计数器，使其可以在lambda中安全修改
            final AtomicInteger processedCount = new AtomicInteger(0);
            final int finalInsertedRowCount = insertedRowCount;
            
            // 批量确认和删除消息
            for (String messageId : messageMap.keySet()) {
                try {
                    // 确认消息处理完成 (XACK)
                    streamCommands.xack(STREAM_KEY, CONSUMER_GROUP, messageId)
                            .subscribe().with(acknowledgeResult -> {
                                        if (acknowledgeResult > 0) {
                                            // 删除已处理的消息
                                            streamCommands.xdel(STREAM_KEY, messageId)
                                                    .subscribe().with(deleteResult -> {
                                                                if (deleteResult > 0) {
                                                                    LOG.infof("Successfully acknowledged and deleted login event: %s", messageId);
                                                                    int currentCount = processedCount.incrementAndGet();
                                                                    if (currentCount == messageMap.size()) {
                                                                        LOG.infof("Batch processing completed: %d events inserted, %d messages acknowledged and deleted",
                                                                                finalInsertedRowCount, messageMap.size());
                                                                    }
                                                                } else {
                                                                    LOG.warnf("Failed to delete login event: %s", messageId);
                                                                }
                                                            },
                                                            deleteThrowable -> {
                                                                LOG.error("Error deleting login event with ID: %s", messageId, deleteThrowable);
                                                            }
                                                    );
                                        } else {
                                            LOG.warnf("Failed to acknowledge login event: %s", messageId);
                                        }
                                    },
                                    acknowledgeThrowable -> {
                                        LOG.error("Error acknowledging login event with ID: %s", messageId, acknowledgeThrowable);
                                    }
                            );
                } catch (Exception e) {
                    LOG.error("Error acknowledging or deleting login event with ID: %s", messageId, e);
                }
            }
        } catch (Exception e) {
            LOG.error("Error in acknowledgeAndDeleteMessages", e);
        }
    }

    private void logFailedMessages(Map<String, StreamMessage<String, String, String>> messageMap) {
        try {
            for (String messageId : messageMap.keySet()) {
                LOG.warnf("Failed to persist login event: %s, message will be reprocessed", messageId);
            }
        } catch (Exception e) {
            LOG.error("Error in logFailedMessages", e);
        }
    }

    public void createConsumerGroupIfNotExists() {
        try {
            // 尝试创建消费者组，如果已经存在会抛出异常，我们忽略这个异常
            streamCommands.xgroupCreate(STREAM_KEY, CONSUMER_GROUP, "0").subscribe().with(
                    result -> {
                        LOG.infof("Created consumer group: %s", CONSUMER_GROUP);
                    },
                    throwable -> {
                        // 消费者组可能已经存在，这是正常的
                        LOG.debugf("Consumer group might already exist: %s", CONSUMER_GROUP);
                    }
            );
        } catch (Exception e) {
            // 捕获所有其他异常
            LOG.error("Error creating consumer group", e);
        }
    }

    // 定时清理已消费的数据（备用方案）
    @Scheduled(every = "1h")
    public void cleanupProcessedEvents() {
        try {
            LOG.infof("Starting cleanup of processed events");
            // 这里可以添加清理逻辑，例如删除过期的消息
            // 注意：需要小心实现，避免删除未处理的消息
            LOG.infof("Finished cleanup of processed events");
        } catch (Exception e) {
            LOG.error("Error during cleanup of processed events", e);
        }
    }
}